package com.asap.demo.ck.handle;

import com.alibaba.fastjson.JSONObject;
import com.asap.demo.ck.orm.DbFetcher;
import com.asap.demo.ck.orm.SqlParameter;
import com.asap.demo.ck.table.JsonField;
import com.asap.demo.ck.table.JsonTable;
import com.asap.demo.ck.util.PropTransformMap;
import com.asap.demo.ck.util.TimeUtil;
import com.asap.demo.model.Event;

import com.asap.demo.utils.PluginContext;
import com.asap.demo.utils.Utils;
import com.asap.rule.StandardEvent;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.yandex.clickhouse.domain.ClickHouseDataType;


import java.io.Serializable;
import java.nio.charset.Charset;
import java.util.*;

public class ProcessForCk  implements Serializable {
    private static final Logger logger = LoggerFactory.getLogger(Process.class);

    private Map<String, String> configMap = null;
    private String version="V3.0 20210506";
    private List<List<Condition>> listCond=null;
    // 新建一个 DbFetcher 实例
    DbFetcher dbFetcher = null;
    // 获取配置文件中的inputTopic
    RuleParse ruleParse = new RuleParse();

    public boolean initialize(PluginContext Context) throws Exception {
        try {
            logger.info( "当前文件的路径为:{}", Context.getConfigPath());

            // 获得所有配置的映射map
            configMap = PropTransformMap.getInstance().readConfigMap(Context.getConfigPath() + "/conf/CK/cfg.properties");
            ruleParse.setConfigMap(configMap);

            logger.info("configMap"+configMap);


            buildFilter(configMap.get("filter_cond"));
            /**
             * 将自定义配置的 json 反序列化
             */
            ruleParse.parseData(Context.getConfigPath() + "/conf/CK/cfg.json");
            // 新建一个 DbFetcher 实例
            dbFetcher = new DbFetcher(configMap);
        }
        catch (Exception e){
            throw e;
        }

        return true;
    }

    public int processMsg(List<StandardEvent> listEvent) throws Exception {
        int result = 0;
        try {
            result = 0;

            long startTime = System.currentTimeMillis();
            result = 0;
            if (logger.isInfoEnabled()) {
                logger.info( version + "listEvent.size():" + listEvent.size());
            }
            List<List<SqlParameter>> ListParam = new ArrayList<>();
            JsonTable table = ruleParse.getTableDefine();
            for (StandardEvent event : listEvent) {
                try {
                    Object parserBody =event.getAllFields().get("rawMsg");
                    // 取出需要做判断的源IP和目的IP字段值
                    String jsonStr = parserBody.toString();
                    if (!filter(jsonStr)) { // 只处理需要匹配到的数据，在为配置条件是，所有数据都处理
                        if (logger.isInfoEnabled()) {
                            logger.info( "jsonStr has filter ");
                        }
                        continue;
                    }


                    if (StringUtils.equals(event.getHeader().get("topic"), configMap.getOrDefault("dlframe.exceptionTopic", "dlExceptionTopic"))) {
                        // 异常数据，需要先执行判断是否已插入，如果插入了,就不需要处理
                        String key = (String) event.getField(table.getTopicPrimaryKey());
                        if (queryData(key)) {
                            if (logger.isInfoEnabled()) {
                                logger.info( "data has in database " + key);
                            }
                            continue;
                        }
                    }

                    ListParam.add(createParam(event, table));
                } catch (Exception e) {
                    logger.error("" , e,"error:");
                    return 1;
                }
            }

            if (ListParam.size() > 0) {
                try {
                    if (logger.isInfoEnabled()) {
                        logger.info( " 入库记录数：" + ListParam.size());
                    }
                    dbFetcher.executeSqlInBatches(ruleParse.getTableInsertSql(), ListParam);
                } catch (Exception e) {
                    logger.error( "insert data to database ", e);
                    result = 1;
                }
            }

            long endTime = System.currentTimeMillis();
            if (logger.isInfoEnabled()) {
                logger.info( version + " 入库耗时：" + (endTime - startTime) + "ms ");
            }
        } catch (Exception e) {
            logger.error( "", e,"error.....");
            return 1;
        }

        return result;
    }

    public boolean notifyFinish() {
        return false;
    }

    public void rollBack() {

    }

    /**
     * 创建 sql 参数
     *
     * @param event 标准事件(kafka message)
     * @param table 自定义配置表
     * @return
     */
    public List<SqlParameter> createParam(StandardEvent event, JsonTable table) {
        List<SqlParameter> params = new ArrayList<SqlParameter>();
        List<JsonField> fields = table.getJsonFieldList();
        // 遍历自定义配置字段集
        for (int i = 0; i < fields.size(); i++) {
            String topicFiledName = fields.get(i).getTopicFieldName();
            String dbFieldName = fields.get(i).getDbFieldName();
            String fieldType = fields.get(i).getFieldType();

            String value = null;
            List<List<String>> extendFieldData = null;

            if (fieldType.contains(ClickHouseDataType.Nested.name()) ) {
                extendFieldData = new ArrayList<>();
                if(event.getField(topicFiledName)!=null){
                    List<List<String>> list = new ArrayList<>();
                    List<String> listTmp = new ArrayList<>();
                    listTmp.add(event.getField(topicFiledName));
                    list.add(listTmp);
                    extendFieldData.addAll(list );
                }else{
                    extendFieldData.add(new ArrayList<>());
                    extendFieldData.add(new ArrayList<>());
                }
            } else {
                value = String.valueOf(event.getField(topicFiledName));
            }

            String resultValue = "";
            if (!fieldType.contains(ClickHouseDataType.Nested.name())) {
                if (value != null && !value.equals("null")) {
                    //处理日期字段中的微秒
                    if (fields.get(i).getFieldType().equals("date") ) {
                        value = Utils.transforDate(value);
                    }
                    resultValue = value;
                }
                else {
                    if (fields.get(i).getDefaultValue() != null) {
                        // 当字段类型为date且默认值为now或者无默认值时
                        if (StringUtils.equals(fieldType, "date"))
                            resultValue = TimeUtil.getCurrentTime();
                        else
                            resultValue = fields.get(i).getDefaultValue();
                    } else {
                        params.add(null);
                    }
                }
            }

            SqlParameter sqlParameter = new SqlParameter().setTopicFieldName(topicFiledName)
                    .setTopicFieldName(dbFieldName)
                    .setFieldType(fieldType)
                    .setFieldValue(resultValue)
                    .setExtendFieldData(extendFieldData);

            params.add(sqlParameter);
        }

        return params;
    }

    private boolean queryData(String keys){

        return queryCkData(keys);

    }

    private boolean queryCkData(String key){
        String sql = ruleParse.getTableQuerySql();
        sql = sql.replace("?", key);
        return  dbFetcher.queryCk(sql);
    }

    private Condition parseCondition(String cond){
        // 在对条件进行解析的时候， 因为匹配的原始数据都是json的，所以在条件上都加了""还有:
        // 比如配置的条件为DEVICE_PARENT_TYPE=IPS, 实际数据中的是 "DEVICE_PARENT_TYPE":"IPS",所以在构建最终用于contain匹配时的条件
        //有将原始配置的条件，加上""，且key,value之间加上了:
        StringBuffer sb = new StringBuffer();
        Condition condition = null;
        if(cond.contains("!=")){
            String[] elems = cond.split("!=");
            sb.append("\"").append(elems[0].trim()).append("\"").append(":\"").append(elems[1].trim()).append("\"");
            condition = new Condition(1,sb.toString());
        }else{
            String[] elems = cond.split("=");
            sb.append("\"").append(elems[0].trim()).append("\"").append(":\"").append(elems[1].trim()).append("\"");
            condition = new Condition(0, sb.toString());
        }

        return condition;
    }

    private List<Condition> parseCondGroups(String condGroups){
        List<Condition> listCondGroups = new ArrayList<>();
        if(StringUtils.isNotEmpty(condGroups)){
            if(condGroups.contains(" AND ")){
                String[] conds = condGroups.split(" AND ");
                for(int i=0; i<conds.length; i++){
                    if(StringUtils.isNotEmpty(conds[i])) {
                        listCondGroups.add(parseCondition(conds[i]));
                    }
                }
            }else{
                listCondGroups.add( parseCondition(condGroups));
            }
        }

        return listCondGroups;
    }

    private void buildFilter(String filterCond){
        this.listCond = new ArrayList<>();
        if(StringUtils.isNotEmpty(filterCond)){
           if(filterCond.contains(" OR ")){
               String[] condGroups = filterCond.split(" OR ");
               for(int i=0; i< condGroups.length; i++){
                   listCond.add(parseCondGroups(condGroups[i]));
               }
           }else{
               listCond.add(parseCondGroups(filterCond));
           }
        }

        logger.info("listCond.size:"+listCond.size());
        logger.info("listCond"+listCond.toString());
    }

    private  boolean matchCond(Condition cond, String json){
        boolean flag = true;
        if(cond.getType()==0){ // 0 正常匹配
            if (!json.contains(cond.getCondition())){  //没匹配上条件 比如 DEVICE_PARENT_TYPE=IPS,数据中是DEVICE_PARENT_TYPE=WAF
               flag = false;
            }
        }else{ //1 取反匹配
            if (json.contains(cond.getCondition())){ //没匹配上条件，比如配置的条件为EVENT_ONE_TYPE!=99, 数据中是 EVENT_ONE_TYPE=99
                flag = false;
            }
        }
        //logger.info("cond"+cond.toString());
        return  flag;
    }

    private boolean matchConds(List<Condition>conds,String json){
        for(Condition cond:conds){
            if(!matchCond(cond,json)){ // 在一个条件组里面，一个条件没匹配上，就是没匹配上
                return false;
            }
        }

        return  true;
    }

    private boolean filter(String json){
       // logger.info("json"+json);
        if(this.listCond.isEmpty()) {
            return true;
        }else{
            for(List<Condition>conds:listCond){
                if(matchConds(conds,json)){ // 有一个条件组匹配，就表示匹配上了
                    return true;
                }
            }
        }
        return false;
    }

}

